package com.fxd.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Component
public class RedisTemplateUtil {

    @Autowired
    private StringRedisTemplate redisTemplate;

    public boolean hashKey(String key) {
        if (StringUtils.isBlank(key)) {
            return false;
        }
        return redisTemplate.hasKey(key);
    }

    /**
     * 使用基数树来索引包含线性数十个stream条目的宏节点。
     * 通常，当你从stream中删除一个条目的时候，条目并没有真正被驱逐，只是被标记为删除。
     *
     * @param stream
     * @param messageId
     * @return 返回成功删除的条目的数量
     */
    public boolean xDel(String stream, RecordId messageId) {
        return redisTemplate.opsForStream().delete(stream, messageId) == 1;
    }

    /**
     * 将指定的流条目追加到指定key的流中。 如果key不存在，作为运行这个命令的副作用，将使用流的条目自动创建key。
     *
     * @param stream
     * @param map
     * @return
     */
    public RecordId xadd(String stream, Map<String, Object> map) {
        return redisTemplate.opsForStream().add(stream, map);
    }

    /**
     * 创建与流关联的新消费者组。
     *
     * @param stream
     * @param groupName
     * @return
     */
    public String xCreateGroup(String stream, String groupName) {
        return redisTemplate.opsForStream().createGroup(stream, groupName);
    }

    /**
     * 只获得与流关联的所有消费者组名称的输出
     *
     * @param stream
     * @return
     */
    public List<String> xInfoGroups(String stream) {
        StreamInfo.XInfoGroups groups = redisTemplate.opsForStream().groups(stream);
        return groups.stream().map(group -> group.groupName()).collect(Collectors.toList());
    }

    /**
     * 只获得与流关联的消费组下所有消费者名称的输出
     *
     * @param stream
     * @return
     */
    public List<String> xInfoConsumers(String stream, String groupName) {
        StreamInfo.XInfoConsumers consumers = redisTemplate.opsForStream().consumers(stream, groupName);
        return consumers.stream().map(consumer -> consumer.consumerName()).collect(Collectors.toList());
    }

    /**
     * 生产者生产一条消息
     *
     * @param stream 生产者流
     * @param object 消息对象
     * @return
     */
    public RecordId xProduce(String stream, Object object) {
        ObjectRecord<String, Object> record = StreamRecords.newRecord()
                .ofObject(object)
                .withStreamKey(stream);
        return redisTemplate.opsForStream().add(record);
    }

    /**
     * 从流的消费者组的待处理条目列表（简称PEL）中删除一条或多条消息。
     *
     * @param stream    流
     * @param groupName 分组名称
     * @param recordId  消息id
     * @return 该命令返回成功确认的消息数
     */
    public Long xAck(String stream, String groupName, RecordId recordId) {
        Long successCount = redisTemplate.opsForStream().acknowledge(stream, groupName, recordId);
        log.info("stream: {},消息id: {},ack结果: {}", stream, recordId, successCount == 1 ? true : false);
        return successCount;
    }

    /**
     * 此命令输出该消费者组的待处理消息的数量（即1），
     * 然后是待处理消息的最小和最大ID，
     * 然后列出消费者组中每一个至少一条待处理消息的消费者，
     * 以及他的待处理消息数量。
     *
     * @param stream    流
     * @param groupName 分组名称
     * @return
     */
    public PendingMessagesSummary xPending(String stream, String groupName) {
        return redisTemplate.opsForStream().pending(stream, groupName);
    }

    /**
     * @param stream    流
     * @param groupName 分组名称
     * @param count     返回消息条数
     * @return 待处理消息列表中有每一条消息的详细信息
     */
    public PendingMessages xPendingInfo(String stream, String groupName, Integer count) {
        return redisTemplate.opsForStream().pending(stream, groupName, Range.unbounded(), count);
    }

    /**
     * @param stream     流
     * @param groupName  分组名称
     * @param toConsumer 将消息转发给目标消费者
     * @param duration   只有在其空闲时间大于我们通过XCLAIM指定的空闲时间的时才会被认领
     * @param recordIds  消息id，可以传多个
     * @return
     */
    public MapRecord<String, String, String> xClaim(String stream, String groupName, String toConsumer, Duration duration, RecordId... recordIds) {
        List<ByteRecord> byteRecords = redisTemplate.execute((RedisConnection conn) ->
                conn.streamCommands().xClaim(stream.getBytes(), groupName, toConsumer, duration, recordIds)
        );
        //反序列化
        List<MapRecord<String, String, String>> mapRecords = new ArrayList<>();
        byteRecords.forEach(byteRecord -> mapRecords.add(byteRecord.deserialize(RedisSerializer.string())));
        return mapRecords.get(0);
    }

}
